SpringBoot 7.1.RabbitMQ进阶使用

本节，我们进一步的了解如何在Spring Boot中使用RabbitMQ。

RabbitMQ基本概念

结构图

mq1

概念说明

Broker：简单来说就是消息队列服务器实体。
Exchange：消息交换机，它指定消息按什么规则，路由到哪个队列。
Queue：消息队列载体，每个消息都会被投入到一个或多个队列。
Binding：绑定，它的作用就是把exchange和queue按照路由规则绑定起来。
Routing Key：路由关键字，exchange根据这个关键字进行消息投递。
vhost：虚拟主机，一个broker里可以开设多个vhost，用作不同用户的权限分离。
producer：消息生产者，就是投递消息的程序。
consumer：消息消费者，就是接受消息的程序。
channel：消息通道，在客户端的每个连接里，可建立多个channel，每个channel代表一个会话任务。

消息队列的使用过程

1.客户端连接到消息队列服务器，打开一个channel。
2.客户端声明一个exchange，并设置相关属性。
3.客户端声明一个queue，并设置相关属性。
4.客户端使用routing key，在exchange和queue之间建立好绑定关系。
5.客户端投递消息到exchange。

exchange接收到消息后，就根据消息的key和已经设置的binding，进行消息路由，将消息投递到一个或多个队列里。

exchange也有几个类型，完全根据key进行投递的叫做Direct交换机，例如，绑定时设置了routing key为abc，那么客户端提交的消息，只有设置了key为abc的才会投递到队列。对key进行模式匹配后进行投递的叫做Topic交换机，符号#匹配一个或多个词，符号*匹配正好一个词（以.来分割）。例如abc.#匹配abc.def.ghi，abc.*只匹配abc.def。还有一种不需要key的，叫做Fanout交换机，它采取广播模式，一个消息进来时，投递到与该交换机绑定的所有队列。

消息的持久化

也就是数据写在磁盘上，为了数据安全考虑，我想大多数用户都会选择持久化。消息队列持久化包括3个部分：
1.exchange持久化，在声明时指定durable = 1
2.queue持久化，在声明时指定durable = 1
3.消息持久化，在投递时指定delivery_mode = 2（1是非持久化）

如果exchange和queue都是持久化的，那么它们之间的binding也是持久化的。如果exchange和queue两者之间有一个持久化，一个非持久化，就不允许建立绑定。

在Spring Boot中实际应用

Direct Exchange

直接匹配模式，通过Exchange名称+RoutingKey来发送与接收消息。

定义一个RoutingKey=”string”的Queue：

@Configuration
public class RabbitConfig {
    @Bean
    public Queue stringQueue() {
        return new Queue("string");
    }
}

向RoutingKey=”string”的Queue投递消息：

@Component
public class HelloProducer {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void sendToStringQueue(String msg) {
        rabbitTemplate.convertAndSend("string", msg);
    }
}

接收消息并消费：

@Component
@RabbitListener(queues = "string")
@Slf4j
public class StringConsumer {

    @RabbitHandler
    public void process(String msg) {
        log.info("string Queue msg : " + msg);
    }
}

跟上节一样，定义这个3类，调用sendToStringQueue(String msg)方法，就可以完成String类型的消息传递了。

如果我们要传递实体类Bean怎么办，很简单，把String改成实体类就可以了：

@Configuration
public class RabbitConfig {
    @Bean
    public Queue userQueue() {
        return new Queue("user");
    }
}

@Component
public class HelloProducer {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void sendToUserQueue(User user) {
        rabbitTemplate.convertAndSend("user", user);
    }
}

@Component
@RabbitListener(queues = "user")
@Slf4j
public class UserConsumer {
    @RabbitHandler
    public void process(User user){
        log.info("user queue msg : " + user);
    }
}

如果我们把User类投递到”string”Queue会怎么样？

1
2
3

public void sendToStringQueue(User user) {
    rabbitTemplate.convertAndSend("string", user);
}

那么该消息会因为找不到对应的消费者而已经在队列中一直等待消费，可以看到我们服务器在疯狂报错：

1
2

org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException: Listener threw exception...
Caused by: org.springframework.amqp.AmqpException: No method found for class com.tt.study.demo.entity.User...

我们可以去RabbitMQ的web监控界面删掉这条消息，也可以再在StringConsumer里定义一个Handler：

@Component
@RabbitListener(queues = "string")
@Slf4j
public class StringConsumer {

    @RabbitHandler
    public void process(String msg) {
        log.info("string Queue msg : " + msg);
    }

    @RabbitHandler
    public void process(User user){
        log.info("user msg : " + user);
    }

}

没错，我们的”string”Queue，既可以消费String类型的消息，又可以消费User类型的消息，这下nb了。

不过我并不推荐这种用法，还是给每种类型的消息都定义一个Queue吧。

Fanout Exchange

广播消息，向所有消费者发布消息，但是只有消费者将队列绑定到该路由器才能收到消息，忽略RoutingKey。

在RabbitConfig类中新建3个Queue和一个FanoutExchange，并把3个Queue和FanoutExchange绑定起来：

@Configuration
public class RabbitConfig {
    ...
    //--------------------------广播--------------------------
    @Bean
    public Queue fanoutAQueue(){
        return new Queue("fanout.a");
    }

    @Bean
    public Queue fanoutBQueue(){
        return new Queue("fanout.b");
    }

    @Bean
    public Queue fanoutCQueue(){
        return new Queue("fanout.c");
    }

    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange("fanoutExchange");
    }

    /**
     * 把fanoutAQueue和fanoutExchange绑定起来
     * @param fanoutAQueue 注意这个参数名必须和你定义Queue的方法名相同
     * @param fanoutExchange 注意这个参数名必须和你定义FanoutExchange的方法名相同
     * @return
     */
    @Bean
    public Binding bindingFanoutAQueue(Queue fanoutAQueue,FanoutExchange fanoutExchange){
        return BindingBuilder.bind(fanoutAQueue).to(fanoutExchange);
    }

    @Bean
    public Binding bindingFanoutBQueue(Queue fanoutBQueue,FanoutExchange fanoutExchange){
        return BindingBuilder.bind(fanoutBQueue).to(fanoutExchange);
    }

    @Bean
    public Binding bindingFanoutCQueue(Queue fanoutCQueue,FanoutExchange fanoutExchange){
        return BindingBuilder.bind(fanoutCQueue).to(fanoutExchange);
    }

}

为3个Queue新建3个对应的Consumer类：

@Component
@RabbitListener(queues = "fanout.a")
@Slf4j
public class FanoutAConsumer {
    @RabbitHandler
    public void process(String msg) {
        log.info("fanout.a msg :", msg);
    }
}

@Component
@RabbitListener(queues = "fanout.b")
@Slf4j
public class FanoutBConsumer {
    @RabbitHandler
    public void process(String msg) {
        log.info("fanout.b msg :", msg);
    }
}

@Component
@RabbitListener(queues = "fanout.c")
@Slf4j
public class FanoutCConsumer {
    @RabbitHandler
    public void process(String msg) {
        log.info("fanout.c msg :", msg);
    }
}

在HelloProducer中定义一个发送消息的方法：

1
2
3

public void sendToFanoutExchange(String msg) {
    rabbitTemplate.convertAndSend("fanoutExchange", null, msg);
}

调用这个方法，发现3个Consumer都消费了这个消息。

Topic Exchange

主题匹配订阅，这里的主题指的是RoutingKey，RoutingKey可以采用通配符，如*或#，RoutingKey命名采用.来分隔多个词，只有消息这将队列绑定到该路由器且指定RoutingKey符合匹配规则时才能收到消息。

在RabbitConfig类中再新建3个Queue和一个TopicExchange，并把Queue和TopicExchange根据不同的RoutingKey绑定：

@Configuration
public class RabbitConfig {
    ...
    //--------------------------主题--------------------------
    @Bean
    public Queue topicAQueue() {
        return new Queue("topic.a");
    }

    @Bean
    public Queue topicBQueue() {
        return new Queue("topic.b");
    }

    @Bean
    public Queue topicCQueue() {
        return new Queue("topic.c");
    }

    @Bean
    public TopicExchange topicExchange() {
        return new TopicExchange("topicExchange");
    }

    @Bean
    public Binding bindingTopicAQueue(Queue topicAQueue, TopicExchange topicExchange) {
        return BindingBuilder.bind(topicAQueue).to(topicExchange).with("key.a");
    }

    @Bean
    public Binding bindingTopicBQueue(Queue topicBQueue, TopicExchange topicExchange) {
        return BindingBuilder.bind(topicBQueue).to(topicExchange).with("key.*");
    }

    @Bean
    public Binding bindingTopicBCQueue(Queue topicCQueue, TopicExchange topicExchange) {
        return BindingBuilder.bind(topicCQueue).to(topicExchange).with("key.#");
    }

}

再定义3个Queue的Consumer：

@Component
@RabbitListener(queues = "topic.a")
@Slf4j
public class TopicAConsumer {
    @RabbitHandler
    public void process(String msg) {
        log.info("topic.a msg : {}", msg);
    }
}

@Component
@RabbitListener(queues = "topic.b")
@Slf4j
public class TopicBConsumer {
    @RabbitHandler
    public void process(String msg) {
        log.info("topic.b msg : {}", msg);
    }
}

@Component
@RabbitListener(queues = "topic.c")
@Slf4j
public class TopicCConsumer {
    @RabbitHandler
    public void process(String msg) {
        log.info("topic.c msg : {}", msg);
    }
}

再在HelloProducer中定义3个生产消息的方法：

public void sendWithTopicA(String msg) {
    rabbitTemplate.convertAndSend("topicExchange", "key.a", msg);
}
public void sendWithTopicB(String msg) {
    rabbitTemplate.convertAndSend("topicExchange", "key.b", msg);
}
public void sendWithTopicBC(String msg) {
    rabbitTemplate.convertAndSend("topicExchange", "key.b.c", msg);
}

我们调用这个3个方法测试一下：

@Test
public void test4() {
    String msg = "topic msg ";
    helloProducer.sendWithTopicA(msg + "1");
    helloProducer.sendWithTopicB(msg + "2");
    helloProducer.sendWithTopicBC(msg + "3");
    log.info("ok");
}

console输出：

...TopicAConsumer      : topic.a msg : topic msg1
...TopicBConsumer      : topic.b msg : topic msg1
...TopicCConsumer      : topic.c msg : topic msg1
...TopicBConsumer      : topic.b msg : topic msg2
...TopicCConsumer      : topic.c msg : topic msg2
...TopicCConsumer      : topic.c msg : topic msg3

我们分析一下：

1.name=”topic.a”的topicAQueue绑定了topicExchange并且RoutingKey=”key.a”，消费者为TopicAConsumer类。
2.name=”topic.b”的topicBQueue绑定了topicExchange并且RoutingKey=”key.*“，消费者为TopicBConsumer类。3.name=”topic.c”的topicCQueue绑定了topicExchange并且RoutingKey=”key.#”，消费者为TopicCConsumer类。

当我们向topicExchange发送RoutingKey=”key.a”的消息时，topicAQueue、topicBQueue、topicCQueue都符合匹配规则，所以3个Consumer类都能消费消息。当发送RoutingKey=”key.b”的消息时，topicAQueue不符合匹配规则，所以无法消费消息。当发送RoutingKey=”key.b.c”的消息时，只有topicCQueue还符合匹配规则，所有只有TopicCConsumer消费到消息。

Headers Exchange

消息头订阅，消息发布前，为消息定义一个或多个键值对的消息头，然后消费者接收消息同时需要定义类似的键值对请求头:(如:x-mactch=all或者x_match=any)，只有请求头与消息头匹配，才能接收消息，忽略RoutingKey。

这种方式用到的比较少（懒），我就不写了。

修改默认的序列化方式

RabbitTemplate默认的MessageConverter为org.springframework.amqp.support.converter.SimpleMessageConverter，如果要传递实体类，需要实体类实现Serializable接口。我们可以使用Jackson2JsonMessageConverter来代替原来的MessageConverter，这样实体就不需要实现Serializable接口，而且在RabbitMQ的Console页面可以更直观的显示消息体内容。

修改方法很简单，只要在RabbitConfig中加入下面方法：

//修改默认的序列化方式为jackson序列化方式
@Bean
public MessageConverter messageConverter() {
    return new Jackson2JsonMessageConverter();
}

在实际业务中的应用

异步处理

例子：用户注册时，需要处理用户数据存到数据库，存成功后需要发邮件或短信提示用户注册成功，最后页面返回注册成功。实际上，只要数据库保存成功，我们就认为用户注册成功了，我们并不关心邮件短信是否发送成功。所以可以把发邮件发短信这一步，放到消息队列里处理，这样用户感受到注册结果的反应时间变快了。

应用解耦

例子：一个商城项目，有订单系统，库存系统。传统做法，用户下单后，订单系统直接调库存系统的接口扣库存，当库存系统出现故障时（并非库存不足无法扣，而是宕机或其他原因），用户下单失败。这其实是我们不愿意看到的，在库存足够的情况下，我们应该确保用户可以下单成功。这里我们就可以引入消息队列，用户下单后，订单系统完成持久化处理，将消息写入消息队列，返回用户订单下单成功，库存系统获取订单消息，进行库存操作。

流量削峰

例子：秒杀活动，一般在秒杀开始时，流量过大，导致应用挂掉。为解决这个问题，我们需要1.控制活动人数，超过阈值的订单直接丢弃，2.缓解短时间内的高流量。引入消息队列，就能很好的处理这个问题。